package top.lunarye.support.delayMessage;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import top.lunarye.common.constant.RocketDelayConstants;
import top.lunarye.support.utils.RedisUtils;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Set;

/**
 * 定时任务查询是否有没有被发送的消息
 * @author lunar
 * @since 2022-07-12
 */
@Component
@Slf4j
public class DelayMQRedisTask {

    @Resource
    private RedisUtils redisUtils;

    @Resource
    private DelayMessageProducer producer;

    @Scheduled(cron="0 0/1 * * * ? ")
    public void recommitMessage() throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        log.info("定时任务启动，开始查询是否有没有被发送的延迟消息");
        Set<String> messages = redisUtils.zRange(RocketDelayConstants.REDIS_STORE_ZSET, DateUtil.current());
        if (!messages.isEmpty()) {
            for (String message : messages) {
                Map<String, Object> msgMap = JSON.parseObject(message, Map.class);
                String topic = (String) msgMap.get("topic");
                String tags = (String) msgMap.get("tags");
                byte[] body = ((String) msgMap.get("body")).getBytes(StandardCharsets.UTF_8);
                Message realMessage = new Message(topic, tags, body);
                producer.start();
                SendResult sendResult = producer.send(realMessage);
                log.info("重新发送了未被发送的延迟消息：{}", sendResult.getMsgId());
                redisUtils.zRem(RocketDelayConstants.REDIS_STORE_ZSET, message);
            }
        }
        log.info("本轮定时任务完成");
    }
}
